package com.tyssq.rocketmq.producer;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.env.Environment;

import java.util.ArrayList;
import java.util.List;

/**
 * <p>
 * 生产者
 * </p>
 *
 * @author liaozhenghan
 * @date 2021-05-21 16:43:24
 */
@Slf4j
@Getter
public abstract class BaseProducer extends DefaultMQProducer implements InitializingBean {
    private String produceGroupName;
    private String topicName;
    @Autowired
    private Environment env;

    @Override
    public void afterPropertiesSet() throws Exception {

        String prefix = env.getProperty("rocketmq.prefix");
        if (prefix == null) {
            prefix = "";
        }

        // 获取分组和主题
        String groupName = this.getGroupName();
        String topic = this.getTopic();
        if (groupName == null) {
            log.error("{}：未配置分组", this.getClass().getSimpleName());
            throw new Exception(this.getClass().getSimpleName() + "：未配置分组");
        }
        if (topic == null) {
            log.error("{}：未配置主题", this.getClass().getSimpleName());
            throw new Exception(this.getClass().getSimpleName() + "：未配置主题");
        }

        this.produceGroupName = prefix + groupName;
        this.topicName = prefix + topic;

        log.info("生产者：{}，主题：{}，分组：{}，正在创建---------------------------------------",
                this.getClass().getSimpleName(), topicName, produceGroupName);
        this.setProducerGroup(produceGroupName);
        this.beforeStart();
        this.start();
        log.info("生产者：{}，主题：{}，分组：{}，创建成功---------------------------------------",
                this.getClass().getSimpleName(), topicName, produceGroupName);
    }

    @Override
    @Value("${rocketmq.namesrvAddr}")
    public void setNamesrvAddr(String namesrvAddr) {
        super.setNamesrvAddr(namesrvAddr);
    }

    protected String getGroupName() {
        return getTopic();
    }

    protected abstract String getTopic();

    protected void beforeStart() {

    }

    /**
     * 预设值的延迟时间间隔为：1s、 5s、 10s、 30s、 1m、 2m、 3m、 4m、 5m、 6m、 7m、 8m、 9m、 10m、 20m、 30m、 1h、 2h
     */
    protected void beforeSendMessage(Message msg, String message) {
    }

    public boolean sendMessage(String tag, String keys, String message) {
        Message msg = new Message(topicName, tag, message.getBytes());
        msg.setKeys(keys);
        try {
            beforeSendMessage(msg, message);
            SendResult send = this.send(msg);
            log.info("消息发送成功,keys:{},{}", keys, send);
            return true;
        } catch (Exception e) {
            log.error("消息发送失败,topic:{},keys:{}", topicName, keys, e);
            return false;
        }
    }

    public boolean batchSend(String... messages) {
        List<Message> list = new ArrayList<>();
        for (String message : messages) {
            Message msg = new Message(topicName, "", message, message.getBytes());
            beforeSendMessage(msg, message);
            list.add(msg);
        }
        try {
            SendResult send = this.send(list);
            log.info("消息发送成功,keys:{},{}", messages, send);
            return true;
        } catch (Exception e) {
            log.error("消息发送失败,topic:{},keys:{}", topicName, messages, e);
            return false;
        }
    }
}
